package cas.ihep.alluxio.impl;

import alluxio.Configuration;
import alluxio.PropertyKey;
import alluxio.client.file.FileSystemContext;
import alluxio.client.netty.NettyRPC;
import alluxio.client.netty.NettyRPCContext;
import alluxio.proto.dataserver.Protocol;
import alluxio.util.proto.ProtoMessage;
import alluxio.wire.WorkerNetAddress;
import com.google.common.base.Preconditions;
import io.netty.channel.Channel;

import java.io.Closeable;
import java.io.IOException;

public class AlluxioReadableBlock extends AlluxioBaseBlock {

    private ClosableResource mResource;
    private static final long READ_TIMEOUT_MS =
            Configuration.getMs(PropertyKey.USER_NETWORK_NETTY_TIMEOUT_MS);

    static class ClosableResource implements Closeable{
        WorkerNetAddress address;
        Channel channel;
        FileSystemContext ctx;
        ClosableResource(FileSystemContext c,WorkerNetAddress addr)throws IOException{
            channel=c.acquireNettyChannel(address=addr);
            ctx=c;
        }
        @Override
        public void close() {
            ctx.releaseNettyChannel(address,channel);
        }
    }

    AlluxioReadableBlock(AlluxioFileIn file, WorkerNetAddress address, long id, long size)throws IOException{
        super(id,size);
        try {
            AlluxioBaseSystem baseSystem=(AlluxioBaseSystem) file.getSystem();
            mResource=mCloser.register(new ClosableResource(baseSystem.getContext(),address));
            Protocol.LocalBlockOpenRequest request =
                    Protocol.LocalBlockOpenRequest.newBuilder().setBlockId(id)
                            .setPromote(file.mInOptions.getAlluxioStorageType().isPromote()).build();
            ProtoMessage message = NettyRPC
                    .call(NettyRPCContext.defaults().setChannel(mResource.channel).setTimeout(READ_TIMEOUT_MS),
                            new ProtoMessage(request));
            Preconditions.checkState(message.isLocalBlockOpenResponse());
            mPath = message.asLocalBlockOpenResponse().getPath();
            mClosed=false;
        } catch(IOException e) {
            mCloser.close();
            mClosed=true;
            throw e;
        }
    }

    public void close() throws IOException {
        if (mClosed) {
            return;
        }
        Protocol.LocalBlockCloseRequest request =
                Protocol.LocalBlockCloseRequest.newBuilder().setBlockId(mId).build();
        try {
            NettyRPC.call(NettyRPCContext.defaults().setChannel(mResource.channel).setTimeout(READ_TIMEOUT_MS),
                    new ProtoMessage(request));
        } finally {
            mClosed=true;
            mCloser.close();
        }
    }

    //public String getBlockPath() {
    //    return mPath;
    //}

    @Override
    public boolean canRead() {
        return true;
    }

    @Override
    public boolean canWrite() {
        return false;
    }

    @Override
    public void cancel() throws IOException {
        close();
    }

    @Override
    public void require(long size) {

    }
}
